package com.bw.gmall.realtime.dws.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.commom.bean.TradeSkuOrderBean;
import com.bw.gmall.realtime.commom.constant.Constant;
import com.bw.gmall.realtime.commom.util.HbaseUtil;
import com.bw.gmall.realtime.commom.util.RedisUtil;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

public abstract class AsyncDimFunction <T> extends RichAsyncFunction<T, T> implements IsyncDim<T> {
    private AsyncConnection hBaseAsyncConnection;
    private StatefulRedisConnection<String, String> redisAsyncConnection;
    private String tableName;
    public AsyncDimFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseAsyncConnection = HbaseUtil.getHBaseAsyncConnection();
        redisAsyncConnection = RedisUtil.getRedisAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeAsyncConnection(hBaseAsyncConnection);
        RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);
    }
    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {
        String dimId = getDimId(t);
        String redisKey = RedisUtil.getRedisKey(tableName, dimId);
        CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
            @Override
            public JSONObject get() {

                String result = null;
                RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(redisKey);
                try {
                    result = stringRedisFuture.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
                return JSONObject.parseObject(result);
            }
        }).thenApplyAsync(new Function<JSONObject, JSONObject>() {
            @Override
            public JSONObject apply(JSONObject jsonObject) {
                JSONObject dimSkuInfo =  null;
                if(jsonObject == null){// 如果Redis没有查出来
                    try {
                        // 从Hbase中查
                        dimSkuInfo = HbaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, tableName, dimId);
                        // 放到Redis当中
                        redisAsyncConnection.async().setex(redisKey,24*60*60, JSON.toJSONString(dimSkuInfo));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }else{// 从Redis中查出来了
                    dimSkuInfo  = jsonObject;
                }
                return dimSkuInfo;
            }
        }).thenAccept(new Consumer<JSONObject>() {
            @Override
            public void accept(JSONObject jsonObject) {

                if (jsonObject != null) {
                    getResult(t,jsonObject);
                } else {
                    System.out.println("\"未查找维度\" = " + "未查找维度");
                }
            }
        });
    }


}
